POC: KEDA ScaledJob for queue jobs (payload stays in Redis)#41
Open
abnegate wants to merge 11 commits into
Open
POC: KEDA ScaledJob for queue jobs (payload stays in Redis)#41abnegate wants to merge 11 commits into
abnegate wants to merge 11 commits into
Conversation
Alternative to the env-var KubernetesJob broker: instead of inlining the payload in a per-message Job, producers enqueue to the ordinary Redis broker and a KEDA ScaledJob scales one-shot worker Jobs off the queue depth. Each worker drains the queue with the same Redis broker and exits — no custom broker, no payload on any Kubernetes object. Adds a drain worker, Dockerfile, redis + ScaledJob manifests, a kubectl-driven e2e (KedaTest) proving KEDA spawns Jobs that drain the queue, and keda-e2e.sh (kind + KEDA + redis). Verified end-to-end on kind. See servers/Keda/README.md for the comparison. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a proof-of-concept for running utopia-php/queue jobs as Kubernetes Jobs via KEDA ScaledJob, keeping job payloads in the existing Redis queue (no Kubernetes involvement at enqueue time, no payload stored on K8s objects).
Changes:
- Introduces a KEDA worker image + one-shot drain worker script and a
ScaledJobmanifest for scaling off Redis list depth. - Adds an end-to-end PHPUnit test (
KedaTest) plus akind + KEDA + Redisharness script to validate the flow. - Documents the approach and trade-offs vs the env-var
KubernetesJobapproach.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/queue/tests/Queue/servers/Keda/worker.php | One-shot worker that drains Redis queue messages using the existing Redis broker. |
| packages/queue/tests/Queue/servers/Keda/README.md | Documentation for running the KEDA ScaledJob POC and trade-offs. |
| packages/queue/tests/Queue/servers/Keda/k8s.yaml | Kind-friendly namespace + Redis + KEDA ScaledJob manifest for scaling jobs from Redis depth. |
| packages/queue/tests/Queue/servers/Keda/Dockerfile | Builds the worker container image (PHP + redis extension + project code). |
| packages/queue/tests/Queue/E2E/Adapter/KedaTest.php | E2E test that enqueues directly into Redis and asserts KEDA drains the queue via Jobs. |
| packages/queue/tests/keda-e2e.sh | Harness that provisions kind + KEDA, loads the worker image, applies manifests, and runs the test. |
| packages/queue/phpunit.xml | Registers KedaTest in the e2e testsuite (skips unless KEDA_E2E=true). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Contributor
Greptile SummaryThis PR introduces a KEDA
Confidence Score: 4/5The PR is safe to merge as a POC with one defect in the graceful-shutdown path of the new adapter. The new packages/queue/src/Queue/Adapter/KubernetesJob.php — specifically the Important Files Changed
Prompt To Fix All With AIFix the following 1 code review issue. Work through them one at a time, proposing concise fixes.
---
### Issue 1 of 1
packages/queue/src/Queue/Adapter/KubernetesJob.php:75-81
**`stop()` closes the consumer before in-flight message can commit/reject**
The class docblock states "SIGTERM/SIGINT trigger `stop()` so pod termination **finishes** the in-flight message instead of stranding it in the processing list," but the implementation contradicts this. When SIGTERM fires while `process()` is executing (e.g., between `messageCallback` returning and `commit()` being called), `consumer->close()` runs immediately. `process()` then calls `commit()` on a closed connection, which throws; the inner `catch` falls through to `reject()`, which also throws on the same closed connection; both exceptions are swallowed, and the message is orphaned in the processing list — the very outcome the docblock says this prevents.
The consumer does not need to be closed here; `Server.php` already closes it in the `workerStop` callback (line 335). The `RECEIVE_TIMEOUT = 2` constant means a blocking `receive()` call will unblock within 2 seconds without forcibly closing the connection, which is an acceptable shutdown latency for a Kubernetes Job. Removing the `$this->consumer->close()` call from `stop()` lets the in-flight `process()` complete, then the drain loop exits on the `$this->isStopped()` check, and the consumer is cleaned up by `workerStop`.
Reviews (11): Last reviewed commit: "refactor(queue): fold single-use helpers..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
Factor the kind + KEDA + Redis + ScaledJob provisioning into tests/keda-lib.sh (keda_up/keda_down) and have both keda-e2e.sh and the package's e2e.sh use it, so `bin/monorepo test queue` (and thus CI) stands up KEDA and runs KedaTest against a real cluster instead of skipping it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- keda-lib.sh: pin + SHA-256-verify Helm (same as kind/kubectl) instead of curl|bash of an unpinned installer; only tear down a kind cluster this run created, never a pre-existing one. - KedaTest: also assert the .failed.* list is empty — draining the main queue alone doesn't prove success since receive() pops before handling; note that enqueue() mirrors Redis::enqueue()'s envelope. - k8s.yaml: correct the scaling comment (KEDA scales N Jobs, workers batch-drain) and document the backoffLimit/processing-orphan crash-recovery caveat. - README: crash-recovery section (processing-list reaper via Publisher::retry). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Member
Author
|
Addressed in 4781c8f: greptile-apps
Copilot
|
The KEDA approach still needs one bit of library code: a consumer that drains the queue and exits (so a Job completes) instead of blocking like the Swoole/ Workerman adapters. Extract that out of the test worker into Utopia\Queue\Adapter\KubernetesJob, cover it with a bare-host unit test (KubernetesJobAdapterTest), and have the KEDA worker run it via Server. Producers are unchanged — they still enqueue with any Publisher (e.g. the Redis broker). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
keda_up ran unconditionally under set -e, so a provisioning failure (no Docker, kind download failure) aborted the whole e2e suite before phpunit, skipping the Swoole/Workerman/pool tests. Gate it: required in CI (hard fail so KEDA regressions surface), best-effort locally where KedaTest self-skips. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…d termination and boot failures Register pcntl SIGTERM/SIGINT handlers in the KubernetesJob consume loop so graceful pod termination drains the in-flight message instead of being SIGKILLed (PHP is PID 1 in the Job container, so an unhandled SIGTERM is ignored). Run the workerStart callbacks under try/finally so workerStop (Timer cleanup) always runs. Add runsToCompletion() to Adapter, overridden true in KubernetesJob, and rethrow boot/consume-loop failures from Server::start() for run-to-completion adapters so the Job fails (honouring backoffLimit) rather than exiting 0. Guard the Redis broker claim writes after the pop and requeue the payload on failure so a mid-claim error no longer strands the message. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
abnegate
force-pushed
the
feat/queue-keda-poc
branch
from
July 14, 2026 06:40
d399b5c to
df8e154
Compare
php:8.4-cli-alpine ships without pcntl, so the new SIGTERM handling in Adapter\KubernetesJob threw at startup and every KEDA Job exited non-zero (the run-to-completion rethrow working as intended) before draining, failing KedaTest. Suggest ext-pcntl at the package level. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
pcntl conflicts with Swoole's signalfd, so when Swoole is loaded the drain runs inside a coroutine scheduler with a sibling coroutine on Coroutine\System::waitSignal. pcntl remains only as the non-Swoole fallback. Throwables are captured inside the scheduler and rethrown after run() since exceptions cannot cross coroutine boundaries. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Timers and signal watchers registered by workerStart hooks (e.g. a telemetry Timer::tick) must be created and cleared inside the same coroutine scheduler, or Coroutine\run never returns and the Job never completes. Coroutine::cancel on a waitSignal coroutine also leaves an orphaned signalfd watcher that swallows SIGTERM with no callback, so signals now use Process::signal registered and removed around the drain. Proven against the real cloud worker: empty-queue exit 0 in 2s, SIGTERM mid-drain graceful exit 0, message consumed off the queue. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A mid-claim Redis failure now propagates as before; the run-to-completion adapter surfaces it as a failed Job rather than silently requeueing. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Comment on lines
+75
to
+81
| public function stop(): self | ||
| { | ||
| $this->stopped = true; | ||
| $this->consumer->close(); | ||
|
|
||
| return $this; | ||
| } |
Contributor
There was a problem hiding this comment.
stop() closes the consumer before in-flight message can commit/reject
The class docblock states "SIGTERM/SIGINT trigger stop() so pod termination finishes the in-flight message instead of stranding it in the processing list," but the implementation contradicts this. When SIGTERM fires while process() is executing (e.g., between messageCallback returning and commit() being called), consumer->close() runs immediately. process() then calls commit() on a closed connection, which throws; the inner catch falls through to reject(), which also throws on the same closed connection; both exceptions are swallowed, and the message is orphaned in the processing list — the very outcome the docblock says this prevents.
The consumer does not need to be closed here; Server.php already closes it in the workerStop callback (line 335). The RECEIVE_TIMEOUT = 2 constant means a blocking receive() call will unblock within 2 seconds without forcibly closing the connection, which is an acceptable shutdown latency for a Kubernetes Job. Removing the $this->consumer->close() call from stop() lets the in-flight process() complete, then the drain loop exits on the $this->isStopped() check, and the consumer is cleaned up by workerStop.
Prompt To Fix With AI
This is a comment left during a code review.
Path: packages/queue/src/Queue/Adapter/KubernetesJob.php
Line: 75-81
Comment:
**`stop()` closes the consumer before in-flight message can commit/reject**
The class docblock states "SIGTERM/SIGINT trigger `stop()` so pod termination **finishes** the in-flight message instead of stranding it in the processing list," but the implementation contradicts this. When SIGTERM fires while `process()` is executing (e.g., between `messageCallback` returning and `commit()` being called), `consumer->close()` runs immediately. `process()` then calls `commit()` on a closed connection, which throws; the inner `catch` falls through to `reject()`, which also throws on the same closed connection; both exceptions are swallowed, and the message is orphaned in the processing list — the very outcome the docblock says this prevents.
The consumer does not need to be closed here; `Server.php` already closes it in the `workerStop` callback (line 335). The `RECEIVE_TIMEOUT = 2` constant means a blocking `receive()` call will unblock within 2 seconds without forcibly closing the connection, which is an acceptable shutdown latency for a Kubernetes Job. Removing the `$this->consumer->close()` call from `stop()` lets the in-flight `process()` complete, then the drain loop exits on the `$this->isStopped()` check, and the consumer is cleaned up by `workerStop`.
How can I resolve this? If you propose a fix, please make it concise.
loks0n
reviewed
Jul 15, 2026
| @@ -342,6 +342,10 @@ function (?Message $message, Throwable $th): void { | |||
| foreach ($this->errorHooks as $hook) { | |||
| $hook->getAction()(...$this->getArguments($this->resources(), $hook)); | |||
| } | |||
|
|
|||
| if ($this->adapter->runsToCompletion()) { | |||
Contributor
There was a problem hiding this comment.
is it possible to hide this/ from the adapter interface?
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Alternative POC to the env-var
KubernetesJobbroker (#35), built to compare the two ways of running queue jobs as Kubernetes Jobs.Approach
The K8s-native pattern: the payload stays in the Redis queue, and KEDA scales worker Jobs off the queue depth.
Utopia\Queue\Broker\Redisbroker — no custom broker, no Kubernetes involvement at enqueue time.ScaledJobwatches the queue's Redis list length and spawns one-shot worker Jobs (up tomaxReplicaCount).tests/Queue/servers/Keda/worker.php) drains the queue with the same Redis broker (receive→ handle →commit) and exits.Notably this needs zero library code — just the existing Redis broker + a drain worker + the
ScaledJobmanifest.Proven end-to-end
tests/keda-e2e.shstands up kind + KEDA (helm) + Redis + theScaledJob, loads the worker image, and runsKedaTest, which enqueues messages and asserts KEDA spawns Jobs that drain the queue. Verified on kind (OK (1 test, 4 assertions); observed 8queue-worker-*Jobs draining the queue).bin/monorepo check queuepasses (pint + phpstan + rector).env-var
KubernetesJob(#35) vs KEDAScaledJob(this PR)ARG_MAXlimits, visible in pod spec, duplicated per Podappwrite-labs/php-k8sin the appRecommendation: KEDA is the more correct/standard path (keeps the queue a queue, no payload on etcd, no custom broker); its cost is requiring the KEDA operator in the cluster. See
tests/Queue/servers/Keda/README.md.🤖 Generated with Claude Code